package com.gycode.vpush.websocket;

import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.gycode.vpush.dto.request.SendMessageDTO;
import com.gycode.vpush.dto.request.SocketMessageReqDTO;
import com.gycode.vpush.dto.request.WebSocketSendDTO;
import com.gycode.vpush.dto.request.WebSocketSendUserDTO;
import com.gycode.vpush.vo.UnReadMessageVO;
import com.gycode.vpush.entity.SendMessageUser;
import com.gycode.vpush.mapper.SendMessageUserMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.ObjectUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author PC
 */
@ServerEndpoint(value = "/ws/{userId}")
@Component
@Slf4j
public class WebSocketServer {

    /**
     * concurrent包的线程安全Map，用来存放每个客户端对应的MyWebSocket对象。
     * 若要实现服务端与单一客户端通信的话，可以使用Map来存放，其中Key可以为用户标识
     */
    public static Map<String, Session> webSocketMap = new ConcurrentHashMap<>();

    private static SendMessageUserMapper sendMessageUserMapper;

    @Autowired
    public void setJiaXinService(SendMessageUserMapper sendMessageUserMapper) {
        WebSocketServer.sendMessageUserMapper = sendMessageUserMapper;
    }

    private static final String WS_SUCCESS = "SUCCESS";
    private static final String WS_FAILURE = "FAIL";
    private static final String WS_HEARTBEAT = "HEARTBEAT";
    private static final String WS_UN_READ = "UNREAD";

    /**
     * 建立连接成功
     */
    @OnOpen
    public void onOpen(Session session, @PathParam("userId") String userId) {
        //与某个客户端的连接会话，需要通过它来给客户端发送数据
        webSocketMap.put(userId, session);
        log.info("【websocket消息】 有新的连接，userId {}，总数{}",userId,webSocketMap.size());
        getUnReadMessage(userId);
    }

    /**
     * 连接关闭
     */
    @OnClose
    public void onClose(@PathParam("userId") String userId) {
        webSocketMap.remove(userId);
        log.info("【websocket消息】 连接断开，userId {}，总数{}",userId, webSocketMap.size());
    }

    /**
     * 接收客户端消息
     */
    @OnMessage
    public void onMessage(String message) {
        log.info("【websocket消息】 收到客户端发来的消息：{}", message);
        //更新数据库
        SocketMessageReqDTO socketMessageReqDTO = JSONObject.parseObject(message, SocketMessageReqDTO.class);
        if (WS_SUCCESS.equals(socketMessageReqDTO.getCode())) {
            log.info("【websocket消息】 更新消息状态为已提醒 消息id {} 坐席id {}"
                    , socketMessageReqDTO.getMessageId(), socketMessageReqDTO.getUserId());
            updateMessageUser(socketMessageReqDTO);
        } else if (WS_HEARTBEAT.equals(socketMessageReqDTO.getCode())) {
            log.info("【websocket消息】 心跳 坐席id {}", socketMessageReqDTO.getUserId());
            heartbeat(socketMessageReqDTO.getUserId());
        }
    }

    /**
     * 发生错误时调用
     *
     * @param session
     * @param error
     */
    @OnError
    public void onError(Session session, Throwable error) {
        log.error("发生错误: ",error);
    }

    /**
     * 发送定时消息
     */
    public void sendTimingMessage(List<WebSocketSendUserDTO> webSocketSendUserDTOList) {
        log.info("【websocket消息】 发送消息：{}", JSONObject.toJSONString(webSocketSendUserDTOList));

        for (WebSocketSendUserDTO webSocketSendUserDTO : webSocketSendUserDTOList) {
            Session session = webSocketMap.get(webSocketSendUserDTO.getUserId());
            if (!ObjectUtils.isEmpty(session)) {
                if (session.isOpen()) {
                    try {
                        session.getBasicRemote().sendText(JSONObject.toJSONString(webSocketSendUserDTO));
                    } catch (IOException e) {
                        log.error("发送消息异常",e);
                    }
                } else {
                    log.error("session 已关闭，消息发送失败");
                }
            } else {
                log.error("session 不存在，消息发送失败");
            }
        }
    }

    /**
     * 发送及时消息
     */
    public void sendTimelyMessage(SendMessageDTO sendMessageDTO) {
        WebSocketSendDTO webSocketSendDTO = sendMessageDTO.getWebSocketSendDTO();
        List<String> userIdList = webSocketSendDTO.getUserIdList();
        List<WebSocketSendUserDTO> webSocketSendUserDTOList = new ArrayList<>();
        for (String userId : userIdList) {
            WebSocketSendUserDTO webSocketSendUserDTO = new WebSocketSendUserDTO();
            webSocketSendUserDTO.setMessageId(webSocketSendDTO.getMessageId());
            webSocketSendUserDTO.setContent(sendMessageDTO.getContent());
            webSocketSendUserDTO.setUserId(userId);
            webSocketSendUserDTO.setPath(webSocketSendDTO.getPath());
            webSocketSendUserDTO.setIsPointTime(webSocketSendDTO.getIsPointTime());
            webSocketSendUserDTO.setPointTime(webSocketSendDTO.getPointTime());
            webSocketSendUserDTO.setType(sendMessageDTO.getType());
            webSocketSendUserDTO.setTitle(sendMessageDTO.getTitle());
            webSocketSendUserDTOList.add(webSocketSendUserDTO);
        }
        sendTimingMessage(webSocketSendUserDTOList);
    }


    /**
     * 心跳
     */
    private void heartbeat(String userId) {
        Session session = webSocketMap.get(userId);
        SocketMessageReqDTO messageReqDTO = new SocketMessageReqDTO();
        messageReqDTO.setCode(WS_SUCCESS);
        log.info("【websocket消息】 心跳响应：{}",JSONObject.toJSONString(messageReqDTO));
        try {
            session.getBasicRemote().sendText(JSONObject.toJSONString(messageReqDTO));
        } catch (IOException e) {
            log.error("心跳异常",e);
        }
    }

    /**
     * 获取未读消息
     * @param userId
     */
    private void getUnReadMessage(String userId) {
        log.info("【websocket消息】 查询未读的消息 {}",userId);
        UnReadMessageVO unReadMessageVO = new UnReadMessageVO();
        unReadMessageVO.setPointTime(new Date());
        unReadMessageVO.setUserId(userId);
        List<WebSocketSendUserDTO> unReadMessageUserList = sendMessageUserMapper.getUnReadMessageUserList(unReadMessageVO);
        if(!CollectionUtils.isEmpty(unReadMessageUserList)){
            sendTimingMessage(unReadMessageUserList);
        }

    }

    /**
     * 更新消息状态
     */
    private void updateMessageUser(SocketMessageReqDTO socketMessageReqDTO) {
        log.info("【websocket消息】 更新消息状态：{}", JSONObject.toJSONString(socketMessageReqDTO));
        SendMessageUser messageUser = new SendMessageUser();
        messageUser.setStatus(true);
        sendMessageUserMapper.update(messageUser, new LambdaUpdateWrapper<SendMessageUser>()
                .eq(SendMessageUser::getUserId, socketMessageReqDTO.getUserId())
                .eq(SendMessageUser::getSendMessageId, socketMessageReqDTO.getMessageId()));
    }

}
